package bma.common.langutil.concurrent;

import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于队列的处理线程
 * 
 * @author 关中
 * @since 1.0 (imported from BMA.ESP Project)
 * 
 */
public class ProcessThread extends Thread implements Wakeupable {

	final Logger log = LoggerFactory.getLogger(ProcessThread.class);

	protected final ProcessQueue core;
	protected volatile Future4Task<Boolean> startFuture = null;
	protected volatile Future4Task<Boolean> closeFuture = null;

	public ProcessThread(ProcessQueue p) {
		super();
		this.core = p;
	}

	public ProcessThread(String name, ProcessQueue p) {
		super("aip#" + name);
		this.core = p;
		p.setName(name);
	}

	public ProcessThread(ThreadGroup group, String name, long stackSize,
			ProcessQueue p) {
		super(group, null, "aip#" + name, stackSize);
		this.core = p;
		p.setName(name);
	}

	public ProcessThread(ThreadGroup group, String name, ProcessQueue p) {
		super(group, "api#" + name);
		this.core = p;
		p.setName(name);
	}

	public ProcessQueue getCore() {
		return core;
	}

	public Future<Boolean> startProcess() {
		if (this.startFuture != null)
			return this.startFuture;

		this.startFuture = new FutureSimple<Boolean>();
		this.start();
		return this.startFuture;
	}

	@Override
	public void run() {
		if (log.isInfoEnabled()) {
			log.info("start run");
		}

		boolean r = core.initProcess();
		if (this.startFuture != null) {
			this.startFuture.done(r, null);
		}

		if (!r) {
			log.warn("{} startup fail,stop", core.getName());
			return;
		}
		while (true) {
			if (!core.isAlive()) {
				break;
			}
			try {
				core.process(ProcessQueue.MAX_SLEEP_TIME);
			} catch (Throwable e) {
				log.error("run exception", e);

				UncaughtExceptionHandler ueh = this
						.getUncaughtExceptionHandler();
				if (ueh != null) {
					ueh.uncaughtException(this, e);
				}
			}
		}
		core.shutdown();
	}

	public boolean postIntent(Object intent) {
		return core.postIntent(intent);
	}

	public Future<Boolean> close() {
		if (!core.isAlive()) {
			return new FutureSimple<Boolean>(true);
		}
		if (this.closeFuture != null)
			return this.closeFuture;
		this.closeFuture = new FutureSimple<Boolean>();
		if (log.isInfoEnabled()) {
			log.info("{}: closing", core.getName());
		}
		ProcessIntentClose closeIntent = new ProcessIntentClose(
				this.closeFuture);
		if (!core.postIntent(closeIntent)) {
			throw new IllegalStateException("process already closed");
		}
		return this.closeFuture;
	}

	@Override
	public void wakup() {
		this.core.wakup();
	}
}
